package com.demo.kafka.processor;


import kafka.utils.timer.SystemTimer;
import kafka.utils.timer.TimerTask;
import kafka.utils.timer.TimerTaskEntry;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.utils.Time;

import java.util.Queue;

/**
 * 定时任务处理类
 * @author K
 */
@Slf4j
public class KafkaTimerTaskProcessor implements TimerTask {

    /**
     * 延时时间
     */
    private final long delayMs;

    /**
     * 主题
     */
    private final String topic;

    private TimerTaskEntry timerTaskEntry;
    private final Queue<String> queue;

    private final static SystemTimer SYSTEM_TIMER = new SystemTimer("延时消息时间轮", 1000, 60, Time.SYSTEM.hiResClockMs());

    public KafkaTimerTaskProcessor(String topic, long delayMs, Queue<String> queue) {
        this.topic = topic;
        this.delayMs = delayMs;
        this.queue = queue;
    }

    @Override
    public void run() {
        log.info("执行定时任务，分区：" + topic);
        queue.offer(topic);
    }

    @Override
    public long delayMs() {
        return delayMs;
    }

    @Override
    public TimerTaskEntry kafka$utils$timer$TimerTask$$timerTaskEntry() {
        return timerTaskEntry;
    }

    @Override
    public void kafka$utils$timer$TimerTask$$timerTaskEntry_$eq(TimerTaskEntry timerTaskEntry) {
        this.timerTaskEntry = timerTaskEntry;
    }

    public static SystemTimer getSystemTimer() {
        return SYSTEM_TIMER;
    }
}
